package com.maimao.goods.callback.kafka.callback;

import com.alibaba.fastjson.JSON;
import com.maimao.core.constant.EsIndexs;
import com.maimao.core.model.search.ErrorMessage;
import com.maimao.core.model.search.KfkMessageType;
import com.maimao.goods.client.EsDocClient;
import com.maimao.message.kafka.sender.KafkaSender;
import com.maimao.model.search.DocInput;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author MaoLin Wang
 * @date 2021/1/3 5:52 下午
 */
public class PublishCallback implements ListenableFutureCallback {
    private Logger log = LoggerFactory.getLogger(PublishCallback.class);
    private Long goodsId;
    private AtomicInteger retryCount;
    private final KafkaSender kafkaSender;
    private final String topic;
    private final EsDocClient esDocClient;

    public PublishCallback(String topic, Long goodsId, KafkaSender kafkaSender, EsDocClient esDocClient) {
        this.goodsId = goodsId;
        retryCount = new AtomicInteger(1);
        this.kafkaSender = kafkaSender;
        this.topic = topic;
        this.esDocClient = esDocClient;
    }


    /**
     * 商品发布消息发送失败
     *
     * @param ex
     */
    @Override
    public void onFailure(Throwable ex) {
        //重新发送
        int count = retryCount.getAndIncrement();
        if (count < 2) {
            log.error("商品发布消息发送失败，商品id:{}，重发次数：{}", this.goodsId, count);
            kafkaSender.sendCallBack(topic, this.goodsId + "", this.goodsId + "", this);
        } else {
            log.error("商品发布消息发送失败，商品id:{}，重发次数：{}，进行入库..", this.goodsId, count);
            ErrorMessage errorMsg = new ErrorMessage(this.goodsId, KfkMessageType.PUBLISH_ERROR.getName());
            DocInput doc = DocInput.builder().index(EsIndexs.ERROR_MESSAGE_INDEX)
                    .jsonSource(JSON.toJSONString(errorMsg))
                    .type("_doc")
                    .build();
            esDocClient.createDoc(doc);
        }

    }

    @Override
    public void onSuccess(Object result) {
        log.info("商品发布消息成功{}", result);

    }
}
